package com.duoec.dw.lineage.service.analyser.commend;

import com.duoec.dw.lineage.dto.SqoopCommendArgument;
import com.duoec.dw.lineage.exceptions.LineageAnalyseException;
import com.google.common.collect.Lists;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.List;

/**
 * @author xuwenzhen
 */
@Service
public class TargetDirCommendLineAnalyser extends BaseCommendLineAnalyser {
    public static final String PATH_SPLITTER = "/";
    public static final String ALANYSER_ERROR = "解析目标表失败：";
    public static final String STR_DOT = ".";
    public static final String STR_EQ = "=";
    public static final int INT_TWO = 2;
    @Value("${hive.dir:/user/hive/warehouse}")
    private String baseHiveDiv = "/user/hive/warehouse";

    /**
     * 用于检查的起始字段串，如果命中后就会执行
     *
     * @return 起始字段串
     */
    @Override
    public String getStart() {
        return "target-dir ";
    }

    /**
     * 分析
     * target-dir /user/hive/warehouse/duoec_ap_customer_edw_source.db/s_du_ap_customer_guide_estate_delta/dt=${dt}
     *
     * @param importDto 血缘关系数据
     * @param line      需要处理的行
     */
    @Override
    public void analyse(SqoopCommendArgument importDto, String line) {
        String targetDir = line.substring(getStart().length());
        if (!baseHiveDiv.endsWith(PATH_SPLITTER)) {
            baseHiveDiv += PATH_SPLITTER;
        }
        String subTargetDir = targetDir.substring(baseHiveDiv.length());
        int index = subTargetDir.indexOf(STR_DOT);
        if (index == -1) {
            throw new LineageAnalyseException(ALANYSER_ERROR + line);
        }
        String databaseName = subTargetDir.substring(0, index);
        importDto.setHiveDatabase(databaseName);
        index = subTargetDir.indexOf(PATH_SPLITTER);
        if (index == -1) {
            throw new LineageAnalyseException(ALANYSER_ERROR + line);
        }
        subTargetDir = subTargetDir.substring(index + 1);
        index = subTargetDir.indexOf(PATH_SPLITTER);
        String tableName;
        if (index == -1) {
            tableName = subTargetDir;
        } else {
            tableName = subTargetDir.substring(0, index);
            subTargetDir = subTargetDir.substring(index + 1);
            setPartition(importDto, subTargetDir);
        }

        importDto.setHiveTable(tableName);
    }

    private void setPartition(SqoopCommendArgument args, String subTargetDir) {
        if (StringUtils.isEmpty(subTargetDir)) {
            return;
        }
        String[] partitionArray = subTargetDir.split(STR_EQ);
        if (partitionArray.length > 0) {
            List<String> partitionKeys = args.getPartitionKeys();
            List<String> partitionValues = args.getPartitionValues();
            if (partitionKeys == null) {
                partitionKeys = Lists.newArrayList();
                args.setPartitionKeys(partitionKeys);
            }
            if (partitionValues == null) {
                partitionValues = Lists.newArrayList();
                args.setPartitionValues(partitionValues);
            }
            for (int i = 0; i < partitionArray.length / INT_TWO; i++) {
                String key = partitionArray[i * INT_TWO];
                if (!partitionKeys.contains(key)) {
                    partitionKeys.add(key);
                    partitionValues.add(partitionArray[i * INT_TWO + 1]);
                }
            }
        }
    }
}
